package dyyx;

import java.util.Date;
import java.util.NavigableMap;
import java.util.Set;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

public class HbaseTest {

	private static final String HBASE_MASTER = "cdh116:60000";
	// create 'cdh_hbase_test' , 'cf1' ,'cf2'
	private static final String TABLE = "cdh_hbase_test";
	private static final String HADOOP_HOME = "/Users/dugang/soft/hadoop";

	private static final String UTF8 = "utf8";

	public static void main(String[] args) throws Exception {

		System.out.println("HbaseTest," + new Date());

		System.setProperty("hadoop.home.dir", HADOOP_HOME);

		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "cdh116,cdh117,cdh118");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		conf.set("hbase.master", HBASE_MASTER);
		// conf.set("hadoop.home.dir", HADOOP_HOME);

		Connection connection = ConnectionFactory.createConnection(conf);

		Admin admin = connection.getAdmin();

		HTableDescriptor[] alltable = admin.listTables();
		for (HTableDescriptor hTableDesc : alltable) {
			System.out.println(hTableDesc.getNameAsString());
		}

		//
		HTablePool pool = new HTablePool(conf, 1000);

		HTableInterface table = (HTableInterface) pool.getTable(TABLE);

		String rowKey = null;

		System.out.println("autoFlush=" + table.isAutoFlush());
		String now = new Date() + "";
		String value = null;
		for (int i = 0; i < 10; i++) {

			rowKey = "key" + i;
			Put put = new Put(rowKey.getBytes());
			value = rowKey + ".cf1_col1_" + now;
			put.add("cf1".getBytes(), "col1".getBytes(), value.getBytes());
			value = rowKey + ".cf2_col1_" + now;
			put.add("cf2".getBytes(), "col1".getBytes(), value.getBytes());

			value = rowKey + ".cf2_col2_" + now;
			put.add("cf2".getBytes(), "col2".getBytes(), value.getBytes());
			table.put(put);

		}
		table.flushCommits();

		rowKey = "key" + 0;

		Get getop = new Get(rowKey.getBytes());

		Result result = table.get(getop);

		System.out.println(result);

		System.out.println(getResultMap(result));

		// System.out.println(result.list());

		// System.out.println(result.listCells());

		System.out.println("\n\n scan start \n\n");

		byte[] startRow = "key1".getBytes();
		byte[] stopRow = "key7".getBytes();

		Scan scan = new Scan(startRow, stopRow);
		// s.setFilter(filter);
		ResultScanner rs = table.getScanner(scan);

		for (Result item : rs) {
			System.out.println(getResultMap(item));
		}

		table.close();

	}

	private static NavigableMap<String, NavigableMap<String, String>> getResultMap(Result result) throws Exception {
		if (result == null) {
			return null;
		}

		// result.current();
		// NavigableMap<byte[], NavigableMap<byte[], byte[]>
		// result.getNoVersionMap();
		NavigableMap<byte[], NavigableMap<byte[], byte[]>> rowmap = result.getNoVersionMap();

		Set<byte[]> cfs = rowmap.keySet();
		NavigableMap<byte[], byte[]> colmap = null;
		Set<byte[]> cols = null;

		NavigableMap<String, NavigableMap<String, String>> map = new TreeMap<String, NavigableMap<String, String>>();

		for (byte[] cf : cfs) {
			colmap = rowmap.get(cf);
			cols = colmap.keySet();

			NavigableMap<String, String> colStrMap = new TreeMap<String, String>();
			for (byte[] col : cols) {
				colStrMap.put(new String(col, UTF8), new String(colmap.get(col), UTF8));

			}

			map.put(new String(cf, UTF8), colStrMap);

		}

		return map;
	}

	private static NavigableMap<String, NavigableMap<String, NavigableMap<Long, String>>> getResultMapHasVersion(
			Result result) throws Exception {
		if (result == null) {
			return null;
		}

		// result.current();
		// NavigableMap<byte[], NavigableMap<byte[], byte[]>
		// result.getNoVersionMap();
		NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> rowmap = result.getMap();

		Set<byte[]> cfs = rowmap.keySet();
		NavigableMap<byte[], NavigableMap<Long, byte[]>> colmap = null;
		Set<byte[]> cols = null;
		NavigableMap<Long, byte[]> colVersionMap = null;
		// String colstr = null;
		// String cfstr = null;

		NavigableMap<Long, String> colStrVersionMap = null;

		NavigableMap<String, NavigableMap<Long, String>> cfStrMap = null;

		NavigableMap<String, NavigableMap<String, NavigableMap<Long, String>>> map = new TreeMap<String, NavigableMap<String, NavigableMap<Long, String>>>();

		for (byte[] cf : cfs) {
			colmap = rowmap.get(cf);
			cols = colmap.keySet();
			cfStrMap = new TreeMap<String, NavigableMap<Long, String>>();
			for (byte[] col : cols) {
				colVersionMap = colmap.get(col);
				colStrVersionMap = getColStrVersionMap(colVersionMap);
				if (colStrVersionMap == null || colStrVersionMap.isEmpty()) {
					continue;
				}
				cfStrMap.put(new String(col, UTF8), colStrVersionMap);
			}
			map.put(new String(cf, UTF8), cfStrMap);

		}

		return map;
	}

	private static NavigableMap<Long, String> getColStrVersionMap(NavigableMap<Long, byte[]> colVersionMap)
			throws Exception {
		if (colVersionMap == null || colVersionMap.isEmpty()) {
			return null;
		}
		Set<Long> tsSet = colVersionMap.keySet();

		System.out.println("tsSet.size=" + tsSet.size());

		byte[] value = null;

		NavigableMap<Long, String> map = new TreeMap<>();
		for (Long ts : tsSet) {
			value = colVersionMap.get(ts);
			if (value == null || value.length <= 0) {
				continue;
			}
			map.put(ts, new String(value, UTF8));
		}

		return map;
	}

}
